package com.kqkj.dyrule.engine.functions;

import com.kqkj.dyrule.engine.beans.*;
import com.kqkj.dyrule.engine.router.NearFarSegmentQueryRouter;
import com.kqkj.dyrule.engine.util.RuleMonitor;
import com.kqkj.dyrule.engine.util.StateDescContainer;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 规则匹配计算核心函数
 *  规则：
 *        触发事件： K事件，事件属性（P2=V1）
 *        画像属性条件：tag87=v2 tag26=v1   hbase
 *        行为次数条件：2022-04-17-当前  事件C[p6=v8,p12=v12] 做过>=2次 clickhouse
 */
public class RuleMatchKeyedProcessFunction extends KeyedProcessFunction<String, EventBean, RuleMatchResult> {

    NearFarSegmentQueryRouter nearFarQueryRounter;
    ListState<EventBean> beansState;
    @Override
    public void open(Configuration parameters) throws Exception {
        beansState = getRuntimeContext().getListState(StateDescContainer.getEventBeansDesc());
        nearFarQueryRounter = new NearFarSegmentQueryRouter(beansState);

    }

    @Override
    public void processElement(EventBean eventBean,Context context, Collector<RuleMatchResult> collector) throws Exception {
        beansState.add(eventBean);
        //获取规则
        RuleConditions rule = RuleMonitor.getRule();

        boolean isMatch = nearFarQueryRounter.ruleMatch(rule, eventBean, beansState);
        if (!isMatch) return;


        RuleMatchResult ruleMatchResult = new RuleMatchResult(eventBean.getDeviceId(), rule.getRuleId(), eventBean.getTimeStamp(), System.currentTimeMillis());
        System.out.println("匹配规则满足条件"+ruleMatchResult);
        collector.collect(ruleMatchResult);

    }
}
